package com.neusoft.databus.core.job;

import java.util.Date;
import java.util.List;
import java.util.concurrent.atomic.AtomicLong;
import java.util.stream.Collectors;

import org.quartz.DisallowConcurrentExecution;
import org.quartz.JobExecutionContext;
import org.quartz.JobExecutionException;
import org.slf4j.MDC;
import org.springframework.scheduling.quartz.QuartzJobBean;

import com.neusoft.bizcore.common.cache.CacheService;
import com.neusoft.databus.common.bean.CommandBean;
import com.neusoft.databus.common.constant.DatabusConstant;
import com.neusoft.databus.common.dto.GatewayDTO;
import com.neusoft.databus.common.dto.MetricGroupDTO;
import com.neusoft.databus.common.dto.ResourceDTO;
import com.neusoft.databus.core.converter.CommandConverter;
import com.neusoft.databus.core.service.ExecuteCollectService;
import com.neusoft.databus.core.service.GatewayService;
import com.neusoft.databus.core.service.JobExecutableDetermineService;
import com.neusoft.databus.core.service.ResourceService;

import lombok.Getter;
import lombok.Setter;
import lombok.extern.slf4j.Slf4j;

/**
 * Quartz任务定义，继承QuartzJobBean
 *
 * @author sunchf
 * @date 2018年12月26日 上午10:57:20
 */
@Slf4j
@Setter
@DisallowConcurrentExecution
public class CollectMetricJob extends QuartzJobBean {
    private GatewayService gatewayService;
    private ResourceService resourceService;
    private JobExecutableDetermineService jobExecutableDetermineService;
    private ExecuteCollectService executeCollectService;
    private CommandConverter commandConverter;
    private CacheService cacheService;

    @Getter
    public final static AtomicLong sendTotal = new AtomicLong(0);

    @Override
    protected void executeInternal(final JobExecutionContext context) throws JobExecutionException {

        if (!this.jobExecutableDetermineService.canExecute()) {
            CollectMetricJob.log.info("system is uninitialized，discard this job");
            return;
        }

        final long start = System.currentTimeMillis();
        try {
            final String jobGroup = context.getJobDetail().getKey().getGroup();
            final String jobName = context.getJobDetail().getKey().getName();
            final boolean isBatchedJob = jobName.equals(jobGroup);
            if (CollectMetricJob.log.isDebugEnabled()) {
                CollectMetricJob.log.debug("job is triggered, it is a {} job, prepare for essential information",
                        isBatchedJob ? "batch" : "single");
            }
            CommandBean command = null;
            if (this.cacheService.isInCache(DatabusConstant.CACHE_NAME_COMMAND, jobName)) {
                command = (CommandBean) this.cacheService.get(DatabusConstant.CACHE_NAME_COMMAND, jobName);
            } else {
                if (isBatchedJob) {
                    command = this.generateFromGateway(context, jobGroup);
                } else {
                    command = this.genenrateFromResource(context, jobName);
                }
                if (null == command) {
                    return;
                }
                this.cacheService.put(DatabusConstant.CACHE_NAME_COMMAND, jobName, command);
            }

            command.setBeginTime(new Date());
            MDC.put("trace", command.generateTraceId());
            this.sendCommand(command);
            CollectMetricJob.log.info("job sent over，{} resources, elapsed {} ms", command.getResourceBean().size(),
                    System.currentTimeMillis() - start);
            MDC.clear();
        } catch (final Throwable e) {
            CollectMetricJob.log.error(e.getMessage(), e);
        }
    }

    protected void sendCommand(CommandBean commandBean) {
        CollectMetricJob.sendTotal.getAndAdd(1);
        try {
            if (CollectMetricJob.log.isDebugEnabled()) {
                CollectMetricJob.log.debug("job contain {} metrics",
                        commandBean.getResourceBean().stream().mapToInt(it -> it.getMetrics().size()).sum());
            }
            this.executeCollectService.collect(commandBean, commandBean.getGatewayIdentifier());

        } catch (final Exception e) {
            CollectMetricJob.log.error(e.getMessage(), e);
        }
    }

    private CommandBean generateFromGateway(JobExecutionContext context, String gatewayCode) {
        final GatewayDTO gateway = this.gatewayService.findByCode(gatewayCode);
        if (gateway.isDisabled() || !gateway.isBatched()) {
            CollectMetricJob.log.warn(
                    "gateway {} has changed to be disable or be not batched, cancel this batch job once.",
                    gatewayCode);

            return null;
        }
        final List<ResourceDTO> resources = this.resourceService.findByGateway(gateway.getCode()).stream()
                .filter(it -> {
                    final MetricGroupDTO mg = it.getCollectedMetricGroup();
                    return !it.isDisabled() && (null != mg) && (mg.getMetricSize() > 0);
                }).collect(Collectors.toList());
        if (resources.size() == 0) {
            CollectMetricJob.log.warn("not any eligible resource bound to gateway {}, cancel this batch job once.",
                    gatewayCode);
            return null;
        }
        return this.commandConverter.toCommand(resources);
    }

    private CommandBean genenrateFromResource(JobExecutionContext context, String resourceCode) {
        final ResourceDTO resourceDTO = this.resourceService.findByCode(resourceCode);
        if (resourceDTO.isDisabled() || (null == resourceDTO.getGateway()) || resourceDTO.getGateway().isDisabled()
                || resourceDTO.getGateway().isBatched()) {
            CollectMetricJob.log.warn("resource {} has changed, cancel this single job once.", resourceCode);
            return null;
        }
        final MetricGroupDTO mg = resourceDTO.getCollectedMetricGroup();
        if ((null == mg) || (mg.getMetricSize() == 0)) {
            CollectMetricJob.log.warn("not any metric bound to resource {}, cancel this single job once.",
                    resourceCode);
            return null;
        }
        return this.commandConverter.toCommand(resourceDTO);
    }
}
